package spark.example

import org.apache.spark.{SparkConf,SparkContext}

object optimizeMergeFile {
    def main(args: Array[String]) {
        val conf = new SparkConf().setAppName("mergeFile1")

        val sc = new SparkContext(conf)
        //val output = "hdfs://master:9000/spark_mergefile_output"
        val output = args(1).toString
        sc.wholeTextFiles(args(0))
	     .map(x => x._2)
	     .map(x => x.replaceAll("[^\u4e00-\u9fa5]+"," "))
	     .zipWithIndex()
	     .map(x => (1000 + x._2).toString + "\t" + x._1)
	     .repartition(1)
	     .saveAsTextFile(output)
    }
}
/*
map(x => x._2) sc.wholeTextFiles() 创建 RDD,带有文件名，第一个字段是文件名，(file_name,file_context),这样自动实现了，一个文件一行
map(x => x.replaceAll("[^\u4e00-\u9fa5]+"," ")) 过滤非中文的字符
zipWithIndex() 这个将产生索引 (line_context,index) index : Int 从 0 开始自增加 1
map(x => "100" + x._2.toString + "\t" + x._1) 将 index 放前面，转换成 100* 
map(x => (1000 + x._2).toString + "\t" + x._1) 上一行有些问题，索引大于100后变成 100100 
这显然不是我们想要的。我们要的是 1001,1002，...，1507.而且上一行的索引还会混乱，不知道为什么
repartition(1) 以一个分区存储，这样保证所有的文件都存储在一个文件中，这样就达到了我们目的
*/